/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package bgu.sonar.util.buffers;

import bgu.sonar.util.ex.UncheckedIOException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.openide.util.Exceptions;

/**
 * this structure is not thread safe!
 *
 * @author bennyl
 */
public class BlockingByteBuffer {

    private LinkedBlockingQueue<byte[]> buf = new LinkedBlockingQueue<>();
    private byte[] currentRead = null;
    private int curReadIdx = 0;
    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private OutputStream outRedirection = null;

    public int available() {
        rwLock.readLock().lock();
        try {
            int sum = 0;
            if (currentRead != null) {
                sum += currentRead.length - curReadIdx;
            }

            for (byte[] b : buf) {
                sum += b.length;
            }
            return sum;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    /**
     * thread safe!
     *
     * @return
     */
    public byte[] readAvailable() {
        rwLock.readLock().lock();
        try {
            LinkedList<byte[]> safeAll = new LinkedList<>();
            buf.drainTo(safeAll);
            int size = (currentRead != null ? currentRead.length - curReadIdx : 0);
            for (byte[] a : safeAll) {
                size += a.length;
            }

            byte[] all = new byte[size];
            int from = (currentRead != null ? currentRead.length - curReadIdx : 0);
            if (currentRead != null) {
                System.arraycopy(currentRead, curReadIdx, all, 0, from);
            }

            for (byte[] a : safeAll) {
                System.arraycopy(a, 0, all, from, a.length);
                from += a.length;
            }

            currentRead = null;
            return all;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public byte[] read(int amount, long timeout) throws InterruptedException, TimeoutException {
        byte[] ret = new byte[amount];
        read(ret, 0, amount, timeout);
        return ret;
    }

    public int read(byte[] all, int offset, int len, long timeout) throws InterruptedException, TimeoutException {
        rwLock.readLock().lock();
        try {
            int pos = offset;
            int readAmount = 0;

            while (readAmount < len) {

                if (currentRead != null && currentRead.length == curReadIdx) {
                    currentRead = null;
                    curReadIdx = 0;
                }

                if (currentRead == null) {
                    currentRead = buf.poll(timeout, TimeUnit.MILLISECONDS);
                    curReadIdx = 0;
                }

                if (currentRead == null) {
                    throw new TimeoutException("received timeout when waiting to get data from the buffer");
                }

                try {
                    for (; curReadIdx < currentRead.length && readAmount < len; curReadIdx++, pos++) {
                        all[pos] = currentRead[curReadIdx];
                        readAmount++;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) {
                    System.out.println("Got Error: curReadIdx was (after inc) " + curReadIdx + " while currentRead length is " + currentRead.length);
                    throw ex;
                }
            }

            return readAmount;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public byte read() throws InterruptedException {
        try {
            return read(-1);
        } catch (TimeoutException ex) {
            throw new InterruptedException("this cannot happened!");
        }
    }

    public byte read(long timeout) throws InterruptedException, TimeoutException {
        rwLock.readLock().lock();
        try {
            if (currentRead != null && currentRead.length == curReadIdx) {
                currentRead = null;
                curReadIdx = 0;
            }

            if (currentRead == null) {
                if (timeout >= 0) {
                    currentRead = buf.poll(timeout, TimeUnit.MILLISECONDS);
                } else {
                    currentRead = buf.poll();
                }
                curReadIdx = 0;
            }

            if (currentRead == null) {
                throw new TimeoutException("received timeout when waiting to get data from the buffer");
            }

            try {
                return currentRead[curReadIdx++];
            } catch (ArrayIndexOutOfBoundsException ex) {
                System.out.println("Got Error: curReadIdx was (after inc) " + curReadIdx + " while currentRead length is " + currentRead.length);
                throw ex;
            }
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public void write(byte[] data) {
        if (outRedirection != null) {
            try {
                outRedirection.write(data);
            } catch (IOException ex) {
                throw new UncheckedIOException(ex);
            }
            return;
        }

        rwLock.writeLock().lock();
        try {
            if (data.length == 0) {
                return;
            }

            try {
                buf.put(data);
            } catch (InterruptedException ex) {
                Logger.getLogger(BlockingByteBuffer.class.getName()).log(Level.SEVERE, null, ex);
                //this is a linked blocking queue - this exception should never be thrown
            }
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    public void write(ByteBuffer bb) {

        byte[] data = new byte[bb.remaining()];
        if (outRedirection != null) {
            try {
                outRedirection.write(data);
            } catch (IOException ex) {
                throw new UncheckedIOException(ex);
            }
            return;
        }

        bb.get(data);
        write(data);
    }

    public InputStream wrappingInputStream(final long defaultReadTimeout) {
        return new InputStream() {
            @Override
            public int read() throws IOException {
                try {
                    return BlockingByteBuffer.this.read(defaultReadTimeout);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                    Thread.currentThread().interrupt();
                    return -1;
                } catch (TimeoutException ex) {
                    throw new IOException("got timeout while attempt to read");
                }
            }

            @Override
            public int read(byte[] b) throws IOException {
                try {
                    return BlockingByteBuffer.this.read(b, 0, b.length, defaultReadTimeout);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                    Thread.currentThread().interrupt();
                    return -1;
                } catch (TimeoutException ex) {
                    throw new IOException("got timeout while attempt to read");
                }
            }

            @Override
            public int read(byte[] b, int off, int len) throws IOException {
                try {
                    return BlockingByteBuffer.this.read(b, off, len, defaultReadTimeout);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                    Thread.currentThread().interrupt();
                    return -1;
                } catch (TimeoutException ex) {
                    throw new IOException("got timeout while attempt to read");
                }
            }

            @Override
            public int available() throws IOException {
                return BlockingByteBuffer.this.available();
            }
        };
    }

    public OutputStream wrappingOutputStream() {
        return new OutputStream() {
            @Override
            public void write(int b) throws IOException {
                BlockingByteBuffer.this.write(new byte[]{(byte) b});
            }

            @Override
            public void write(byte[] b) throws IOException {
                BlockingByteBuffer.this.write(b);
            }

            @Override
            public void write(byte[] b, int off, int len) throws IOException {
                byte[] cut = new byte[len];
                System.arraycopy(b, off, cut, 0, len);
                BlockingByteBuffer.this.write(cut);
            }
        };
    }

    public void redirectOutput(OutputStream out) {
        try {
            this.outRedirection = out;
            out.write(readAvailable());
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }
}
